package com.ls.fw.data.hbase.dao;

import java.io.IOException;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ls.fw.data.bean.Column;
import com.ls.fw.data.bean.Record;
import com.ls.fw.data.bean.Table;
import com.ls.fw.data.dao.DaoAdapter;
import com.ls.fw.data.enums.FieldType;
import com.ls.fw.data.exception.DataException;
import com.ls.fw.data.hbase.server.CacheClient;
import com.ls.fw.data.hbase.utils.DataTypeConvertUtil;
import com.ls.fw.data.hbase.utils.SynchronizdDateUtil;
import com.ls.fw.data.mapper.ScanRowMapper;
import com.ls.fw.data.result.DataResult;
import com.ls.fw.data.result.DataResultSet;
import com.ls.fw.data.result.DataResultSetImpl;
import com.ls.fw.data.statement.Delete;
import com.ls.fw.data.statement.Insert;
import com.ls.fw.data.statement.Select;
import com.ls.fw.data.statement.Update;
import com.ls.fw.data.visitor.FieldVisitor;

/**
 * Hbase数据服务实现
 * 
 * @author ls 2015年2月12日下午5:33:40
 */
public class DataDaoHbaseImpl extends DaoAdapter {
	
	private static final int PAGESIZE = 10000;

	public static final String FAMILY = "data";
	public static final byte[] FAMILY_BYTE = Bytes.toBytes(FAMILY);

	public static final String SYS_FIELDS = "fields";
	public static final String SYS_INFO = "info";

	public static final byte[] SEQUENCE_BYTE = Bytes.toBytes("sequence");
	public static final String IDSEQUENCE = "idsequence";

	public static final byte[] SYS_INFO_BYTE = Bytes.toBytes(SYS_INFO);
	public static final byte[] SYS_FIELDS_BYTE = Bytes.toBytes(SYS_FIELDS);

	public static final String SYSTABLE = "sysconfig";

	public static final String VERSION = "version";
	public static final String CREATEDATE = "createDate";
	public static final String REMARK = "remark";
	public static final String AUTOPRIVATEKEY = "autoPrivateKey";

	public static final String ID = "id";
	public static final byte[] ID_BYTE = Bytes.toBytes(ID);

	/**
	 * 数据最大版本数
	 */
	public static final int MAX_VERSION = 1;

	protected Logger logger = LoggerFactory.getLogger(this.getClass());

	private HConnection connection;

	private Configuration configuration;

	/**
	 * 缓存
	 */
	private CacheClient cache = new CacheClient();

	public DataDaoHbaseImpl(boolean openCache, String cacheAddress,
			String clientPort, String quorum, String master) {
		cache.setOpen(openCache);
		cache.setAddress(cacheAddress);
		configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", clientPort);
		configuration.set("hbase.zookeeper.quorum", quorum);
		configuration.set("hbase.master", master);
	}

	public DataDaoHbaseImpl(String clientPort, String quorum, String master) {
		this(false, "", clientPort, quorum, master);
	}

	@Override
	public boolean create(Table table) {
		boolean flag = false;
		if (table == null) {
			return false;
		}
		long start = System.currentTimeMillis();
		HBaseAdmin hBaseAdmin = null;
		try {
			hBaseAdmin = new HBaseAdmin(configuration);
			if (hBaseAdmin.tableExists(table.getName())) {
				throw new DataException("表【" + table.getName() + "】已经存在！");
			}
			if (!hBaseAdmin.tableExists(SYSTABLE)) {
				HTableDescriptor tableDescriptor = new HTableDescriptor(
						TableName.valueOf(SYSTABLE));
				tableDescriptor.addFamily(new HColumnDescriptor(SYS_INFO)
						.setMaxVersions(MAX_VERSION));
				tableDescriptor.addFamily(new HColumnDescriptor(SYS_FIELDS)
						.setMaxVersions(MAX_VERSION));
				hBaseAdmin.createTable(tableDescriptor);
			}
			HTableDescriptor tableDescriptor = new HTableDescriptor(
					TableName.valueOf(table.getName()));
			tableDescriptor.addFamily(new HColumnDescriptor(FAMILY)
					.setMaxVersions(MAX_VERSION));
			hBaseAdmin.createTable(tableDescriptor);
			Insert insert = this.creatTableData(table);
			this.insertTableData(insert);
			this.creatTableSequence(table);
			flag = true;
			cache.put("table_" + table.getName(), table);
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		} finally {
			if (null != hBaseAdmin) {
				try {
					hBaseAdmin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		this.logger.info("创建表" + table.getName() + "耗时："
				+ (System.currentTimeMillis() - start) + " ms");
		return flag;
	}

	private void creatTableSequence(Table table) {
		HTableInterface hTableInterface = null;
		try {
			hTableInterface = getConnection().getTable(IDSEQUENCE);
			Put put = new Put(Bytes.toBytes(table.getName()));
			put.add(SEQUENCE_BYTE, ID_BYTE, Bytes.toBytes(0l));
			hTableInterface.put(put);
		} catch (Exception e) {
			throw new DataException(e.getMessage() + "创建表主键自增序列异常");
		} finally {
			if (hTableInterface != null) {
				try {
					hTableInterface.close();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 封装表结构信息
	 * 
	 * @author ls 2015年3月2日下午2:19:55
	 * @param table
	 * @return
	 */
	private Insert creatTableData(Table table) {
		Insert insert = new Insert();
		insert.setTable(SYSTABLE);
		final Record record = insert.startRecord();
		record.addField(ID, table.getName());

		Column dateColumn = new Column(CREATEDATE, CREATEDATE);
		dateColumn.setValue(SynchronizdDateUtil.getDateFormat().format(
				table.getDate()));
		dateColumn.setFieldType(FieldType.date);
		record.addField(dateColumn);

		Column verColumn = new Column(VERSION, VERSION);
		verColumn.setValue(table.getVersion());
		verColumn.setFieldType(FieldType.longx);
		record.addField(verColumn);

		record.addField(REMARK, table.getRemark());

		Column priColumn = new Column(AUTOPRIVATEKEY, table.isAutoGenerateKey());
		priColumn.setFieldType(FieldType.booleanx);
		record.addField(priColumn);

		final AtomicInteger i = new AtomicInteger();
		table.accept(new FieldVisitor() {

			@Override
			public void visit(Column field) {
				i.addAndGet(1);
				record.addField("col" + i.get() + "_name", field.getName());
				record.addField("col" + i.get() + "_indexName",
						field.getIndexName());
				record.addField("col" + i.get() + "_isPrivateKey",
						field.isPrimaryKey());
				record.addField("col" + i.get() + "_isNull", field.getIsNull());
				record.addField("col" + i.get() + "_fieldType", field
						.getFieldType().toString());
				record.addField("col" + i.get() + "_fieldLength",
						field.getFieldLength());
				record.addField("col" + i.get() + "_isCreateIndex",
						field.isIndex());
				record.addField("col" + i.get() + "_sequence", i.get());
				record.addField("col" + i.get() + "_desc", field.getDesc());

			}
		});
		return insert;
	}

	/**
	 * 添加表结构信息
	 * 
	 * @author ls 2015年3月2日下午2:19:35
	 * @param insert
	 * @return
	 */
	private boolean insertTableData(Insert insert) {
		boolean flag = false;
		if (insert == null) {
			return false;
		}
		long start = System.currentTimeMillis();
		HTableInterface hTableInterface = null;
		try {
			hTableInterface = getConnection().getTable(
					insert.getTable().getName());
			List<Record> records = insert.getRecords();
			for (Record record : records) {
				Column idColumn = record.getField(getPrikeyName());
				if (idColumn == null || idColumn.getValue() == null
						|| String.valueOf(idColumn.getValue()).equals("")) {
					throw new DataException("表主键不存在！");
				}
				String id = idColumn.getValue() + "";
				Put put = new Put(Bytes.toBytes(id));
				List<Column> fields = record.getFields();
				for (Column field : fields) {
					if (field.getName().indexOf("col") > -1) {
						put.add(SYS_FIELDS_BYTE,
								Bytes.toBytes(field.getName()),
								DataTypeConvertUtil.dataConvertByte(
										field.getValue(), field.getFieldType()));
					} else {
						put.add(SYS_INFO_BYTE,
								Bytes.toBytes(field.getName()),
								DataTypeConvertUtil.dataConvertByte(
										field.getValue(), field.getFieldType()));
					}
				}
				hTableInterface.put(put);
			}
			this.logger.info("添加" + records.size() + "条记录耗时："
					+ (System.currentTimeMillis() - start) + " ms");
			flag = true;
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		} finally {
			if (hTableInterface != null) {
				try {
					hTableInterface.close();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		return flag;

	}

	@Override
	public boolean drop(Table table) {
		boolean flag = false;
		if (table == null) {
			return false;
		}
		long start = System.currentTimeMillis();
		HBaseAdmin admin = null;
		try {
			admin = new HBaseAdmin(configuration);
			if (admin.tableExists(table.getName())) {
				admin.disableTable(table.getName());
				admin.deleteTable(table.getName());
			}
			//删除表结构
			Delete deleteTableInfo = new Delete();
			deleteTableInfo.setTable(SYSTABLE);
			List<String> infoids = new ArrayList<String>();
			infoids.add(table.getName());
			deleteTableInfo.setIds(infoids);
			this.delete(deleteTableInfo);
			//删除表序列
			Delete deleteTablesequence = new Delete();
			deleteTablesequence.setTable(IDSEQUENCE);
			List<String> sequenceids = new ArrayList<String>();
			sequenceids.add(table.getName());
			deleteTableInfo.setIds(sequenceids);
			this.delete(deleteTablesequence);
			flag = true;
			cache.delete("table_" + table.getName());
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		} finally {
			if (null != admin) {
				try {
					admin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		this.logger.info("删除表" + table.getName() + "耗时："
				+ (System.currentTimeMillis() - start) + " ms");
		return flag;
	}

	@Override
	public boolean update(Table table) {
		return this.create(table);
	}

	private Table assembleTable(String name, Result result) throws Exception {
		Table table = new Table("", new String(result.getRow(), "utf-8"));
		try {
			NavigableMap<byte[], byte[]> infoMap = result
					.getFamilyMap(SYS_INFO_BYTE);
			byte[] date = result.getValue(SYS_INFO_BYTE,
					Bytes.toBytes(CREATEDATE));
			if (date != null && date.length > 0) {
				table.setDate(SynchronizdDateUtil.getDateFormat().parse(
						new String(date)));
			}
			table.setAutoGenerateKey(Bytes.toBoolean(infoMap.get(Bytes
					.toBytes(AUTOPRIVATEKEY))));
			if (infoMap.containsKey(Bytes.toBytes(VERSION))) {
				try {
					table.setVersion(Bytes.toInt(infoMap.get(Bytes
							.toBytes(VERSION))));
				} catch (Exception e) {
				}
			}
			if (infoMap.containsKey(Bytes.toBytes(REMARK))) {
				table.setRemark(new String(infoMap.get(Bytes.toBytes(REMARK)),
						"utf-8"));
			}
			NavigableMap<byte[], byte[]> fieldsMap = result
					.getFamilyMap(SYS_FIELDS_BYTE);
			int i = 0;
			List<Column> fields = new ArrayList<Column>();
			while (true) {
				if (!fieldsMap.containsKey(Bytes.toBytes("col" + i + "_name"))) {
					break;
				}
				Column field = new Column(new String(fieldsMap.get(Bytes
						.toBytes("col" + i + "_name"))));
				if (fieldsMap.containsKey(Bytes.toBytes("col" + i
						+ "_indexName"))) {
					field.setIndexName(new String(fieldsMap.get(Bytes
							.toBytes("col" + i + "_indexName"))));
				}
				if (fieldsMap.containsKey(Bytes.toBytes("col" + i
						+ "_isPrivateKey"))) {
					field.setPrimaryKey(Bytes.toBoolean(fieldsMap.get(Bytes
							.toBytes("col" + i + "_isPrivateKey"))));
				}
				if (fieldsMap.containsKey(Bytes.toBytes("col" + i + "_isNull"))) {
					field.setIsNull(Bytes.toBoolean(fieldsMap.get(Bytes
							.toBytes("col" + i + "_isNull"))));
				}
				if (fieldsMap.containsKey(Bytes.toBytes("col" + i
						+ "_fieldType"))) {
					field.setFieldType(new String(fieldsMap.get(Bytes
							.toBytes("col" + i + "_fieldType"))));
				}
				if (fieldsMap.containsKey(Bytes.toBytes("col" + i
						+ "_fieldLength"))) {
					field.setFieldLength(Bytes.toInt(fieldsMap.get(Bytes
							.toBytes("col" + i + "_fieldLength"))));
				}
				if (fieldsMap.containsKey(Bytes.toBytes("col" + i
						+ "_isCreateIndex"))) {
					field.setIndex(Bytes.toBoolean(fieldsMap.get(Bytes
							.toBytes("col" + i + "_isCreateIndex"))));
				}
				if (fieldsMap.containsKey(Bytes.toBytes("col" + i + "_desc"))) {
					field.setDesc(new String(fieldsMap.get(Bytes.toBytes("col"
							+ i + "_desc"))));
				}
				fields.add(field);
				i++;

			}
			table.setFields(fields);
		} catch (Exception e) {
			throw new DataException(e.getMessage() + ",查询表出错." + e.getCause(), e);
		}
		return table;
	}

	@Override
	public Table getTable(String name) {
		Table t = cache.get("table_" + name);
		if (t != null) {
			return t;
		}
		HTableInterface hTableInterface = null;
		Table table = null;
		HBaseAdmin hBaseAdmin = null;
		long start = System.currentTimeMillis();
		try {
			hBaseAdmin = new HBaseAdmin(configuration);
			if (!hBaseAdmin.tableExists(name)) {
				throw new DataException("表【" + name + "】不存在！");
			}
			hTableInterface = getConnection().getTable(SYSTABLE);
			Get get = new Get(Bytes.toBytes(name));
			Result rs = hTableInterface.get(get);
			table = this.assembleTable(name, rs);
			this.logger.info("获取表" + table.getName() + "耗时："
					+ (System.currentTimeMillis() - start) + " ms");
			cache.put("table_" + table.getName(), table);
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		} finally {
			if (null != hTableInterface) {
				try {
					hTableInterface.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (null != hBaseAdmin) {
				try {
					hBaseAdmin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return table;
	}

	public HConnection getConnection() throws IOException {
		if (connection == null) {
			synchronized (this) {
				connection = HConnectionManager.createConnection(configuration);
			}
		}
		return connection;
	}

	@Override
	public boolean open(Object address) {
		cache.open();
		try {
			connection = this.getConnection();
		} catch (IOException e) {
			throw new DataException(e.getMessage(), e);
		}
		return true;
	}

	@Override
	public boolean commit(boolean softCommit) {
		return true;
	}

	@Override
	public boolean rollback() {
		return true;
	}

	@Override
	public void optimize() {
	}

	@Override
	public boolean insert(Insert insert, boolean checkExist) {
		return this.insert(insert, checkExist, true);
	}
	
	
	protected boolean insert(Insert insert, boolean checkExist, boolean isInsert) {
		boolean flag = false;
		if (insert == null) {
			return false;
		}
		long start = System.currentTimeMillis();
		HTableInterface hTableInterface = null;
		HBaseAdmin hBaseAdmin = null;
		try {
			hBaseAdmin = new HBaseAdmin(configuration);
			if (!hBaseAdmin.tableExists(insert.getTable().getName())) {// 如果存在要创建的表，那么先删除，再创建
				throw new DataException("表【" + insert.getTable().getName()
						+ "】不存在！");
			}

			hTableInterface = getConnection().getTable(
					insert.getTable().getName());
			Table table = insert.getTable();
			if (table.getFields() == null || table.getFields().size() == 0) {
				table = this.getTable(insert.getTable().getName());
			}
			Map<String, Column> map = new HashMap<String, Column>();
			if (table.getFields() != null && table.getFields().size() > 0) {
				for (Column field : table.getFields()) {
					map.put(field.getName(), field);
				}
			}
			List<Record> records = insert.getRecords();
			for (Record record : records) {
				
				Put put = this.createInsertPut(map, record, table, checkExist, isInsert);
				hTableInterface.put(put);
			}
			this.logger.info("添加" + records.size() + "条记录耗时："
					+ (System.currentTimeMillis() - start) + " ms");
			flag = true;
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		} finally {
			if (null != hTableInterface) {
				try {
					hTableInterface.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (null != hBaseAdmin) {
				try {
					hBaseAdmin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return flag;

	}

	private String creatRowkey(String tableName) {
		HTableInterface hTableInterface = null;
		Long idValue = 1L;
		try {
			hTableInterface = getConnection().getTable(IDSEQUENCE);
			idValue = hTableInterface.incrementColumnValue(
					Bytes.toBytes(tableName), SEQUENCE_BYTE, Bytes.toBytes(ID),
					1L);
		} catch (Exception e) {
			throw new DataException(
					e.getMessage() + ",自增表" + tableName + "ID异常！", e);
		} finally {
			if (hTableInterface != null) {
				try {
					hTableInterface.close();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		return String.valueOf(idValue);

	}

	private boolean prikeyIsExist(Table table,String rowkey){
		boolean isexist = false;
		HTableInterface hTableInterface = null;
		try{
			hTableInterface = getConnection().getTable(table.getName());
		    Get get = new Get(Bytes.toBytes(rowkey));
		    Result  result =  hTableInterface.get(get);
		    if(result !=null && !result.isEmpty() && result.getRow()!=null){
		    	isexist = true;
		    }
		}catch(Exception e){
			throw new DataException(e.getMessage(),e);
		}
		return isexist;
	}
	/**
	 * 创建添加数据put
	 * 
	 * @author ls 2015年3月2日下午2:32:49
	 * @param map
	 * @param record
	 * @return
	 * @throws Exception
	 * @throws ParseException
	 */
	private Put createInsertPut(Map<String, Column> map, Record record,
			Table table,boolean checkExist, boolean isInsert) throws Exception, ParseException {
		Column idColumn = record.getField(getPrikeyName());
		String id = "";
		if(idColumn != null){
			id = (String) idColumn.getValue();
		}
		if(!isInsert && StringUtils.isBlank(id) ){
			throw new DataException("主键不能为空");
		}
        if(checkExist){
			if(id!=null && !id.equals("")){
				this.prikeyIsExist(table, id);
			}
		}
		if (idColumn == null || idColumn.getValue().toString().equals("")) {
			if (table.isAutoGenerateKey()) {
				FieldType pkfieldType=table.getPriKey().getFieldType();
				if (pkfieldType == FieldType.longx) {
					id = this.creatRowkey(table.getName());
				} else {
					id = UUID.randomUUID().toString().replaceAll("-", "");
				}
				if (idColumn == null) { 
					idColumn=new Column(this.getPrikeyName(),id).setFieldType(pkfieldType);
					record.addField(idColumn);
				} else {
					idColumn.setValue(id).setFieldType(pkfieldType);
				}
			} else {
				throw new DataException("主键不能为空");
			}
		}else{
			if (table.isAutoGenerateKey() && isInsert){
				if(id != null && !id.equals("")){
					throw new DataException("主键为自增列,不能设主键值");
				}
			}
		}
        if(id==null || id.equals("")){
        	throw new DataException("主键不能为空");
        }
		Put put = new Put(Bytes.toBytes(id));
		List<Column> fields = record.getFields();
		for (Column field : fields) {
			String name = field.getName();
			if (name.equals(this.getPrikeyName())) {
				continue;
			}
			Object value = field.getValue();
			if (map.containsKey(name)) {
				Column nameColumn = map.get(name);
				if ((value == null || value.equals(""))
						&& !nameColumn.getIsNull()) {
					throw new DataException("字段值不能为空！");
				}
				if (FieldType.varchar == (nameColumn.getFieldType())
						&& value != null
						&& value.toString().length() > nameColumn
								.getFieldLength()) {
					throw new DataException(name + "字段数据超过限制长度！");
				}
				if (FieldType.date == (nameColumn.getFieldType())) {
					if (value != null && !"".equals(value)) {
						if (!value.toString()
								.equals(SynchronizdDateUtil.formatDate(value
										.toString()))) {
							throw new DataException(name + "(值为：" + value + ")"
									+ "字段不符合日期格式！");
						}
						put.add(FAMILY_BYTE, Bytes.toBytes(name),
								Bytes.toBytes(value.toString()));
					}
				} else {
					put.add(FAMILY_BYTE,
							Bytes.toBytes(name),
							DataTypeConvertUtil.dataConvertByte(value,
									nameColumn.getFieldType()));
				}
			} else {
				throw new DataException("字段【" + name + "】不存在！");
			}
		}
		return put;
	}

	@Override
	public boolean delete(Delete delete) {
		boolean flag = false;
		if (delete == null) {
			return false;
		}
		long start = System.currentTimeMillis();
		HTableInterface hTableInterface = null;
		HBaseAdmin hBaseAdmin = null;
		try {
			hBaseAdmin = new HBaseAdmin(configuration);
			if (!hBaseAdmin.tableExists(delete.getTable().getName())) {
				throw new DataException("表【" + delete.getTable().getName()
						+ "】不存在！");
			}
			hTableInterface = getConnection().getTable(
					delete.getTable().getName());
			List<org.apache.hadoop.hbase.client.Delete> deletes = new ArrayList<org.apache.hadoop.hbase.client.Delete>();
			for (String rowkey : delete.getIds()) {
				org.apache.hadoop.hbase.client.Delete d1 = new org.apache.hadoop.hbase.client.Delete(
						Bytes.toBytes(rowkey));
				deletes.add(d1);
			}
			int size = deletes.size();
			hTableInterface.delete(deletes);
			this.logger.info("删除" + size + "条记录耗时："
					+ (System.currentTimeMillis() - start) + " ms");
			flag = true;
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		} finally {
			if (null != hTableInterface) {
				try {
					hTableInterface.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (null != hBaseAdmin) {
				try {
					hBaseAdmin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return flag;

	}

	@Override
	public boolean update(Update update, boolean checkExist) {
		try {
				if(!update.hasWhere()){
					Insert insert = new Insert();
					insert.setFromItem(update.getTable());
					insert.setRecords(update.getRecords());
					this.insert(insert,checkExist,false);
			    }else{
					Select select = new Select();
					select.setFromItem(update.getTable());
					select.setIds(update.getIds());
					DataResult sr = this.search(select);
					List<Record> records = sr.getResults();
					for (Record record : records) {
						List<Column> fields = record.getFields();
						List<Column> updateColumns =  update.getRecords().get(0).getFields();
						for (Column field : fields) {
							for (Column updatefield : updateColumns) {
								if(field.getName().equals(updatefield.getName())){
									field.setValue(updatefield.getValue());
								}

							}
						}
					}
					Insert insert = new Insert();
					insert.setFromItem(update.getTable());
					insert.setRecords(records);
					this.insert(insert,checkExist,false);
			}
			return true;
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		}

	}

	@Override
	public boolean exist(Select select) {
		boolean isexist = false;
		HTableInterface hTableInterface = null;
		try{
			hTableInterface = getConnection().getTable(select.getTable().getName());
		    List<Get> gets = new ArrayList<Get>();
		    for(String rowkey : select.getIds()){
		    	Get get = new Get(Bytes.toBytes(rowkey));
		    	gets.add(get);
		    }
		    Boolean[] result =  hTableInterface.exists(gets);
		    for (Boolean boolean1 : result) {
				if(boolean1){
					return boolean1;
				}
			}
		}catch(Exception e){
			throw new DataException(e.getMessage(),e);
		}
		return isexist;
		
	}

	@Override
	public DataResult search(Select select) {
		if (select == null) {
			return null;
		}
		long start = System.currentTimeMillis();
		HTableInterface hTableInterface = null;
		DataResult ser = new DataResult();
		HBaseAdmin hBaseAdmin = null;
		try {
			hBaseAdmin = new HBaseAdmin(configuration);
			if (!hBaseAdmin.tableExists(select.getTable().getName())) {
				throw new DataException("表【" + select.getTable().getName()
						+ "】不存在！");
			}
			hTableInterface = getConnection().getTable(
					select.getTable().getName());
			Table table = this.getTable(select.getTable().getName());
			Map<String, FieldType> map = new HashMap<String, FieldType>();
			if (table.getFields() != null && table.getFields().size() > 0) {
				for (Column field : table.getFields()) {
					map.put(field.getName(), field.getFieldType());
				}
			}
			List<Record> records = new ArrayList<Record>();
			List<String> ids = select.getIds();
			if (select.hasId()) {
				final int size  = ids.size();
				if(PAGESIZE < size){
					for (int i = 0; i < size; i += PAGESIZE) {
						List<Get> lget = new ArrayList<Get>();
						List<String> tempindex = ids.subList(i, (size<(i + PAGESIZE)?size:(i + PAGESIZE)));
						for (String rowkey : tempindex) {
							Get get = new Get(Bytes.toBytes(rowkey));
							lget.add(get);
						}
						Result[] results = hTableInterface.get(lget);
						if (results != null && results.length > 0) {
							for (Result r : results) {
								if(!r.isEmpty()){
									Record record = this.getRecord(r, map);
									records.add(record);
								}
							}
						}
					}
				}else{
					List<Get> lget = new ArrayList<Get>();
					for (String rowkey : ids) {
						Get get = new Get(Bytes.toBytes(rowkey));
						lget.add(get);
					}
					Result[] results = hTableInterface.get(lget);
					if (results != null && results.length > 0) {
						for (Result r : results) {
							if(!r.isEmpty()){
								Record record = this.getRecord(r, map);
								records.add(record);
							}
						}
					}
				}
				
			} else {
				Scan scan = new Scan();
				ResultScanner rs = hTableInterface.getScanner(scan);
				Iterator<Result> results = rs.iterator();
				while (results.hasNext()) {
					Result r = results.next();
					Record record = this.getRecord(r, map);
					records.add(record);
				}
			}
			ser.setResults(records);
			this.logger.info("查询" + records.size() + "条记录耗时："
					+ (System.currentTimeMillis() - start) + " ms");
		} catch (DataException e1) {
			throw e1;
		} catch (Exception e) {
			throw new DataException(e.getMessage(), e);
		} finally {
			if (null != hTableInterface) {
				try {
					hTableInterface.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (null != hBaseAdmin) {
				try {
					hBaseAdmin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return ser;
	}

	private Record getRecord(Result result, Map<String, FieldType> map) {
		Record record = new Record();
		List<Column> fieldList = new ArrayList<Column>();
		fieldList.add(new Column(this.getPrikeyName(), new String(result
				.getRow())));
		NavigableMap<byte[], byte[]> infoMap = result.getFamilyMap(FAMILY_BYTE);
		if (infoMap != null && infoMap.size() > 0) {
			for (byte[] clonmu : infoMap.keySet()) {
				FieldType fieldType = map.get(new String(clonmu));
				if (fieldType != null) {
					byte[] value = infoMap.get(clonmu);
					Column field = new Column(new String(clonmu));
					field.setValue(DataTypeConvertUtil.byteConvertObject(value,
							fieldType));
					fieldList.add(field);
				}
			}
		}
		record.setFields(fieldList);
		return record;
	}

	@Override
	public <T> long scan(Select select, ScanRowMapper<T> rowMapper)
			throws SQLException {
		HTableInterface hTableInterface = null;
		HBaseAdmin hBaseAdmin = null;
		long count = 0;
		try {
			hBaseAdmin = new HBaseAdmin(configuration);
			if (!hBaseAdmin.tableExists(select.getTable().getName())) {
				throw new DataException("表【" + select.getTable().getName()
						+ "】不存在！");
			}
			hTableInterface = getConnection().getTable(
					select.getTable().getName());
			Table table = this.getTable(select.getTable().getName());
			Map<String, FieldType> map = new HashMap<String, FieldType>();
			if (table.getFields() != null && table.getFields().size() > 0) {
				for (Column field : table.getFields()) {
					map.put(field.getName(), field.getFieldType());
				}
			}	
			List<String> ids = select.getIds();
			Scan scan = new Scan();
			if (ids != null && ids.size() > 0) {
				if (ids.size() == 1) {
					scan.setStartRow(Bytes.toBytes(ids.get(0)));
				} else {
					scan.setStartRow(Bytes.toBytes(ids.get(0)));
					scan.setStopRow(Bytes.toBytes(ids.get(ids.size() - 1)));
				} 
			}
			ResultScanner rs = hTableInterface.getScanner(scan);
			Iterator<Result> it = rs.iterator();
			DataResult searchResult = new DataResult();
			List<Record> recordList = new ArrayList<Record>();
			int i = 0;
			int page = 0;
			boolean goon = true;
			while (it.hasNext() && goon) {
				count++;
				if (i < select.getRowCount()) {
					Result result = it.next();
					if (rowMapper.showField(page + 1)) {
						Record record = this.getRecord(result, map);
						recordList.add(record);
					}
					i++;
				} else if (i == select.getRowCount()) {
					i = 0;
					page++;
					searchResult.setResults(recordList);
					goon = this.handResult(rowMapper, searchResult, page, goon);
					recordList.clear();
				}
			}

		} catch (Exception e) {
			throw new DataException(e.getMessage());
		} finally {
			if (null != hTableInterface) {
				try {
					hTableInterface.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (null != hBaseAdmin) {
				try {
					hBaseAdmin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}

		return count;
	}

	private <T> boolean handResult(ScanRowMapper<T> rowMapper,
			DataResult searchResult, int page, boolean goon)
			throws SQLException {
		List<T> list;
		if (rowMapper != null) {
			DataResultSet resultSet = new DataResultSetImpl(searchResult);
			list = new ArrayList<T>(resultSet.getFetchSize());
			while (resultSet.next()) {
				T obj = rowMapper.mapRow(resultSet,1);
				if (obj != null) {
					list.add(obj);
				}
			}
			resultSet.close();
		} else {
			list = (List<T>) searchResult.getResults();
		}
		if (rowMapper != null) {
			goon = rowMapper.processingOfResults(list, page);
		}
		return goon;
	}

	@Override
	public void close() {
		if (null != connection) {
			try {
				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public Object getServer() {
		return connection;
	}

	@Override
	public long load(Table fromTable, Select select, Table toTable,
			boolean deleteOldData, ScanRowMapper<Record> rowMapper)
			throws SQLException {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public long executeQueryForInt(Select select) throws SQLException {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public void execute(String sql) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void execute(String sql, boolean checkExist) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public long executeQueryForInt(String sql) throws SQLException {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public List<?> executeQuery(String sql) {
		// TODO Auto-generated method stub
		return null;
	}

}
